use std::{
  collections::BTreeMap,
  sync::{
    atomic::{AtomicU32, Ordering},
    OnceLock,
  },
  time::SystemTime,
};

use futures::StreamExt;
use pulsar::{
  error::ConnectionError, producer, proto, Consumer, Producer, ProducerOptions, Pulsar, SubType,
  TokioExecutor,
};
use tokio::{
  runtime::Runtime,
  sync::{mpsc, oneshot},
};
use tracing::{debug, error, warn};

use self::call_return::{CallBackReceiver, CallResultRegister};

pub(crate) mod call_return;

static PULSAR_CLIENT: OnceLock<PulsarWrapper> = OnceLock::new();
static CURR_CALL_ID: AtomicU32 = AtomicU32::new(0);
const CALL_FUN_TOPIC: &str = "callFunction";

///调用方法时的结构
#[derive(Debug)]
pub struct CallFunMsg {
  pub fun_name: &'static str,
  pub call_id: i32,
  pub msg: String,
}

impl CallFunMsg {
  pub fn new(fun_name: &'static str, msg: String, call_id: i32) -> Self {
    Self {
      fun_name,
      call_id,
      msg,
    }
  }

  fn to_message(self, key: String) -> producer::Message {
    let payload = format!(
      r#"{{"_params":{},"_fun_name":"{}","_call_id":{}}}"#,
      self.msg, self.fun_name, self.call_id
    )
    .into_bytes();

    producer::Message {
      payload,
      partition_key: Some(key),
      ..Default::default()
    }
  }
}

///向发送守护协程发送的消息内容
#[derive(Debug)]
enum SendMessage {
  ///直接的消息
  DirectlyMsg(&'static str, String),
  ///调用方法的消息.
  CallFunMsg(CallFunMsg),
  ///增加一个发布者
  AddProducer(&'static str),
}

///pulsar的一个包装器,用来在整个程序当中保证只有唯一一个实例.
pub struct PulsarWrapper {
  client: Pulsar<TokioExecutor>,
  // producers: BTreeMap<String, pulsar::Producer<TokioExecutor>>,
  send_msg_sender: mpsc::UnboundedSender<SendMessage>,
  call_back_sender: mpsc::UnboundedSender<CallResultRegister>,
  key: String,
}

impl PulsarWrapper {
  pub fn new(addr: String, key: String, rt: Runtime) -> Result<&'static Self, pulsar::Error> {
    Ok(PULSAR_CLIENT.get_or_init(|| {
      rt.block_on(async move {
        let client = match Pulsar::builder(addr, TokioExecutor).build().await {
          Ok(r) => r,
          Err(e) => panic!("pulsar client init error:{}", e),
        };

        let (send_msg_sender, send_msg_receiver) = mpsc::unbounded_channel();
        let call_back_sender = match CallBackReceiver::start_running(
          format!("{CALL_FUN_TOPIC}_{key}"),
          key.clone(),
          &client,
        )
        .await
        {
          Ok(r) => r,
          Err(e) => panic!("call back receiver init error:{}", e),
        };

        tokio::spawn(start_runing_pulsar(
          client.clone(),
          key.clone(),
          send_msg_receiver,
        ));
        let wrapper = Self {
          client,
          key,
          send_msg_sender,
          call_back_sender,
        };

        wrapper
      })
    }))
  }

  ///向指定的topic发送消息
  pub fn send(&self, topic: &'static str, msg: String) {
    let _ = self
      .send_msg_sender
      .send(SendMessage::DirectlyMsg(topic, msg));
  }

  ///增加一个消息队列发送器
  pub fn add_producer(&self, topic: &'static str) {
    let _ = self.send_msg_sender.send(SendMessage::AddProducer(topic));
  }

  ///给一个消息接收对象..可以根据此接收器进行消息的接收操作
  pub async fn consumer(
    &self,
    topic_name: &'static str,
    subscription_type: SubType,
  ) -> Result<Consumer<String, TokioExecutor>, pulsar::Error> {
    self
      .client
      .consumer()
      .with_topic(topic_name)
      .with_consumer_name(self.key.clone())
      .with_subscription_type(subscription_type)
      .with_subscription("test_subscription")
      .build()
      .await
  }

  ///调用远程函数,并阻塞等待返回
  pub async fn call_function(
    &self,
    fun_name: &'static str,
    para: String,
    timeout_millisecond: u64,
  ) -> Result<String, pulsar::Error> {
    let call_id = CURR_CALL_ID.fetch_add(1, Ordering::Relaxed) as i32;
    let msg = CallFunMsg::new(fun_name, para, call_id);

    let (sender, mut r) = oneshot::channel::<String>();
    let reg = CallResultRegister { call_id, sender };
    //向得到返回值协程注册
    let _ = self.call_back_sender.send(reg);
    //发消息出去
    let _ = self.send_msg_sender.send(SendMessage::CallFunMsg(msg));

    let current_time = SystemTime::now();
    loop {
      //超时判断
      let now = SystemTime::now();
      if let Ok(v) = now.duration_since(current_time) {
        if v.as_millis() as u64 > timeout_millisecond {
          return Err(pulsar::Error::Custom("timeout".to_string()));
        }
      }

      //正常返回
      if let Ok(res) = r.try_recv() {
        return Ok(res);
      }
    }
  }

  ///收一个消息,此方法需要一个Consume对象,在此基础上处理相应方法.
  /// 并将得到的数据返回处理
  pub async fn consume_message(
    consumer: &mut Consumer<String, TokioExecutor>,
  ) -> Result<String, pulsar::Error> {
    match consumer.next().await {
      Some(Ok(msg)) => match msg.deserialize() {
        Ok(s) => {
          consumer.ack(&msg).await?;
          Ok(s)
        }
        Err(e) => {
          error!("反序列化消息失败:{}", e);
          Err(pulsar::Error::Custom(format!("{e}")))
        }
      },
      Some(Err(e)) => {
        error!("接收消息失败:{}", e);
        Err(e)
      }
      None => {
        error!("接收消息失败:{}", "消息为空");
        Err(pulsar::Error::Connection(ConnectionError::Disconnected))
      }
    }
  }

  /// 当收到消息即调用给定的方法处理消息
  /// 注意当前的执行是在本处理协程当中进行,所以如果时长较长,需要自行增加协程调用
  pub async fn consume_message_run<F>(
    &self,
    topic_name: &'static str,
    subscription_type: SubType,
    f: F,
  ) -> Result<(), pulsar::Error>
  where
    F: Fn(&str, String) -> () + Send + 'static,
  {
    let mut consumer = self
      .client
      .consumer()
      .with_topic(topic_name)
      .with_consumer_name(self.key.clone())
      .with_subscription_type(subscription_type)
      .with_subscription("test_subscription")
      .build::<String>()
      .await?;

    tokio::spawn(async move {
      while let Some(Ok(msg)) = consumer.next().await {
        if let Err(e) = consumer.ack(&msg).await {
          error!("确认消息失败:{}", e);
        }

        match msg.deserialize() {
          Ok(data) => {
            debug!("收到消息,topic:{}开始调用方法:{}", msg.topic, data);
            f(&msg.topic, data);
          }
          Err(e) => {
            error!("反序列化数据失败:{}", e)
          }
        }
      }
    });

    Ok(())
  }
}

///开始启动一个pulsar消息队列对接的方法
/// 此方法实际操作与pulsar的发送
async fn start_runing_pulsar(
  pulsar: Pulsar<TokioExecutor>,
  key: String,
  mut send_msg_receiver: mpsc::UnboundedReceiver<SendMessage>,
) {
  let mut producers = BTreeMap::new();
  let mut call_function = create_producer(&pulsar, CALL_FUN_TOPIC, key.clone())
    .await
    .expect("create producer error");

  loop {
    match send_msg_receiver.recv().await {
      Some(SendMessage::AddProducer(topic_name)) => {
        if let Some(p) = create_producer(&pulsar, topic_name, key.clone()).await {
          producers.insert(topic_name, p);
        }
      }
      Some(SendMessage::DirectlyMsg(topic_name, msg)) => {
        let msg = producer::Message {
          payload: msg.into_bytes(),
          partition_key: Some(key.clone()),
          ..Default::default()
        };

        match producers.get_mut(topic_name) {
          Some(p) => {
            if let Err(e) = p.send_non_blocking(msg).await {
              error!("向{topic_name}发送消息失败: {}", e);
            }
          }
          None => {
            warn!("没有生产者:{topic_name},新建此生产者,并发送");
            if let Some(mut p) = create_producer(&pulsar, topic_name, key.clone()).await {
              if let Err(e) = p.send_non_blocking(msg).await {
                error!("向{topic_name}发送消息失败: {}", e);
              }
              producers.insert(topic_name, p);
            }
          }
        };
      }
      Some(SendMessage::CallFunMsg(msg)) => {
        let _ = call_function.send_non_blocking(msg.to_message(key.clone()));
      }
      None => {
        error!("消息发送者已经全部退出,退出当前协程");
        return;
      }
    };
  }
}

#[inline]
async fn create_producer(
  pulsar: &Pulsar<TokioExecutor>,
  topic_name: &str,
  key: String,
) -> Option<Producer<TokioExecutor>> {
  match pulsar
    .producer()
    .with_topic(topic_name)
    .with_name(key)
    .with_options(ProducerOptions {
      schema: Some(proto::Schema {
        r#type: proto::schema::Type::String as i32,
        ..Default::default()
      }),

      ..Default::default()
    })
    .build()
    .await
  {
    Ok(p) => Some(p),
    Err(e) => {
      error!("新建生产者失败:{e}");
      None
    }
  }
}
